package com.messaging;

import com.messaging.events.MessageNotSentEvent;
import com.messaging.events.MessageReceiveEvent;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.Deque;
import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author Dmitry
 */
public class UdpMq {

    private final DatagramSocket server;
    private final DatagramSocket sender;
    private final BlockingQueue<Message> messages;
    private Thread ackProcessThread;
    private Thread messageSendThread;
    private Thread messageReceiveThread;
    private int ackIndex = 1;
    private MessageReceiveEvent messageReceiveEvent;
    private MessageNotSentEvent messageNotSentEvent;
    private final Deque<WaitAckMessage> waitAckMessages = new ConcurrentLinkedDeque<WaitAckMessage>();
    private final Deque<ProcessedPacket> processedAcks = new ConcurrentLinkedDeque<>();
    private final int RESEND_PACKET_INTERVAL = 5000;
    private final int MAX_RETRY_COUNT = 25;
    private final int PROCESSED_ACK_PACKET_EXIST_TIME = 120000;
    private int sendsReliable = 0;
    private int resendsReliable = 0;

    public UdpMq(int port) {
        messages = new LinkedBlockingQueue<>();
        try {
            server = new DatagramSocket(port);
            server.setSoTimeout(500);
            sender = new DatagramSocket();
        } catch (SocketException ex) {
            throw new RuntimeException(ex);
        }
    }

    public int getMessageQueuedCount() {
        return messages.size();
    }

    public int waitAckMessagesCount() {
        return waitAckMessages.size();
    }

    public int getSendsReliable() {
        return sendsReliable;
    }

    public int getResendsReliable() {
        return resendsReliable;
    }

    public void setMessageReceiveEvent(MessageReceiveEvent event) {
        this.messageReceiveEvent = event;
    }

    public void setMessageNotSentEvent(MessageNotSentEvent messageNotSentEvent) {
        this.messageNotSentEvent = messageNotSentEvent;
    }

    public void stop() {
        messageSendThread.interrupt();
        messageReceiveThread.interrupt();
        ackProcessThread.interrupt();
        server.close();
        sender.close();
    }

    public void start() {
        messageSendThread = new Thread() {

            @Override
            public void run() {
                while (!isInterrupted()) {
                    queueSender(this);
                }
            }
        };
        messageSendThread.start();

        messageReceiveThread = new Thread() {

            @Override
            public void run() {
                while (!isInterrupted()) {
                    try {
                        messageReceiver(this);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        };
        messageReceiveThread.start();

        ackProcessThread = new Thread() {

            @Override
            public void run() {
                while (!isInterrupted()) {
                    try {
                        processAckStep();
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
            }
        };
        ackProcessThread.start();
    }

    public void sendMessage(byte[] value, InetAddress address, int port) {
        try {
            byte[] bytes = new byte[value.length + 3];
            bytes[0] = 0;//0 at highest bit - packet without ACK
            copyShortToArray(bytes, (short) server.getLocalPort(), 1);//set sender port for return
            System.arraycopy(value, 0, bytes, 3, value.length);
            Message message = new Message(bytes, address, port);
            messages.put(message);
        } catch (InterruptedException ex) {
            throw new RuntimeException(ex);
        }
    }

    public void sendMessageReliable(byte[] value, InetAddress address, int port) {
        try {
            byte[] bytes = new byte[value.length + 6];
            int newAck = ackIndex;
            ackIndex++;
            int untouchedAck = newAck;
            newAck = newAck | 0x8000_0000;//set highest bit to 1 -> packet with ACK
            newAck = newAck & 0xBFFFFFFF;//set second highest bit to 0 -> it is not answer ACK
            copyIntToArray(bytes, newAck, 0);
            copyShortToArray(bytes, (short) server.getLocalPort(), 4);//set sender port for return
            System.arraycopy(value, 0, bytes, 6, value.length);
            Message message = new Message(bytes, address, port);
            WaitAckMessage wam = new WaitAckMessage(untouchedAck, message);
            waitAckMessages.addLast(wam);
            messages.put(message);
            sendsReliable++;
        } catch (InterruptedException ex) {
            throw new RuntimeException(ex);
        }
    }
    int counter = 0;
    int messagePerSeconds = 0;
    int currentMessagePerSeconds = 0;
    long stTime;

    private void queueSender(Thread thread) {
        try {
            stTime = System.currentTimeMillis();
            while (!thread.isInterrupted()) {
                Message message = null;
                try {
                    message = messages.poll(500, TimeUnit.MILLISECONDS);
                } catch (InterruptedException ex) {
                }
                if (message != null) {
                    DatagramPacket packet = new DatagramPacket(message.getValue(), message.getValue().length, message.getAddress(), message.getPort());
                    try {
                        if (!sender.isClosed()) {
                            sender.send(packet);

                            if ((counter % 4) == 0) {
                                System.out.println("Sent" + counter);
                                Thread.sleep(1);
                            }
                            counter++;
                        }
                    } catch (IOException ex) {
                        throw new RuntimeException(ex);
                    }
                }
                long currentTime = System.currentTimeMillis();
                messagePerSeconds++;
                if ((currentTime - stTime) >= 1000) {
                    currentMessagePerSeconds = messagePerSeconds;
                    messagePerSeconds = 0;
                    stTime = currentTime;
                }
            }
        } catch (InterruptedException ex) {
        }
    }

    public int getCurrentMessagePerSeconds() {
        return currentMessagePerSeconds;
    }

    private int getSenderPortFromAddress(DatagramPacket packet) {
        byte[] bytes = packet.getData();
        if ((bytes[0] & 0x80) != 0) {
            return (int) (copyShortFromArray(bytes, 4) & 0xFFFF);
        } else {
            return (int) (copyShortFromArray(bytes, 1) & 0xFFFF);
        }
    }

    private void messageReceiver(Thread thread) throws InterruptedException {
        byte[] buffer = new byte[65535];
        DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
        while (!thread.isInterrupted()) {
            boolean messageReceived = false;
            try {
                server.receive(packet);
                messageReceived = true;
            } catch (SocketTimeoutException | SocketException ex) {
            } catch (IOException ex) {
                ex.printStackTrace();
            }
            if (messageReceived) {
                int port = getSenderPortFromAddress(packet);
                byte[] realBuffer = null;
                if ((buffer[0] & 0xC0) == 0xC0) {//is ACK packet
                    processAckPacket(packet.getAddress(), port, buffer);
                } else if ((buffer[0] & 0x80) != 0) {//the message requires ack response
                    if (processPacketThatRequireAck(packet.getAddress(), port, buffer)) {
                        realBuffer = new byte[packet.getLength() - 6];
                        System.arraycopy(buffer, 6, realBuffer, 0, packet.getLength() - 6);
                    }
                } else {//simple not reliable message
                    realBuffer = new byte[packet.getLength() - 3];
                    System.arraycopy(buffer, 3, realBuffer, 0, packet.getLength() - 3);
                }

                if (messageReceiveEvent != null && realBuffer != null) {
                    try {
                        messageReceiveEvent.messageReceive(realBuffer, packet.getAddress(), port);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }
    }

    private void processAckPacket(InetAddress address, int port, byte[] buffer) {
        int packetAckIndex = copyIntFromArray(buffer, 0) & 0x3FFF_FFFF;
        for (Iterator<WaitAckMessage> iter = waitAckMessages.iterator(); iter.hasNext();) {
            WaitAckMessage m = iter.next();
            if (m.getAckIndex() == packetAckIndex) {
                iter.remove();//ack was received. No need to track it anymore
                break;
            }
        }
    }

    //this packet may be "resend because ack packet was lost". In this case we should send ack and not process it in application level
    private boolean processPacketThatRequireAck(InetAddress address, int port, byte[] buffer) throws InterruptedException {
        int packetAckIndex = copyIntFromArray(buffer, 0) & 0x3FFF_FFFF;
        int senderAddressInt = copyIntFromArray(address.getAddress(), 0);
        int currentTimeMillis = (int) System.currentTimeMillis() & 0x7FFFFFFF;
        sendAckResponse(address, port, buffer);
        for (ProcessedPacket packet : processedAcks) {
            if (packet.getAckIndex() == packetAckIndex && packet.getAddressFrom() == senderAddressInt) {
                return false;
            }
        }

        ProcessedPacket ackPacket = new ProcessedPacket(senderAddressInt, port, packetAckIndex, currentTimeMillis);
        processedAcks.add(ackPacket);
        return true;
    }

    public int getWaitAcksCount() {
        return waitAckMessages.size();
    }

    private void processAckStep() throws Exception {
        int currentTimeMillis = (int) System.currentTimeMillis() & 0x7FFFFFFF;
        for (Iterator<ProcessedPacket> iter = processedAcks.iterator(); iter.hasNext();) {
            ProcessedPacket packet = iter.next();
            int timeDifference = currentTimeMillis - packet.getTime();
            if (timeDifference > 0) {
                if (timeDifference > PROCESSED_ACK_PACKET_EXIST_TIME) {
                    iter.remove();
                }
            } else if (timeDifference < 0) {
                timeDifference = packet.getTime() - currentTimeMillis;
                if (timeDifference > PROCESSED_ACK_PACKET_EXIST_TIME) {
                    iter.remove();
                }
            }
        }

        long currentTime = System.currentTimeMillis();
        for (Iterator<WaitAckMessage> iter = waitAckMessages.iterator(); iter.hasNext();) {
            WaitAckMessage packet = iter.next();
            if ((currentTime - packet.getLastRetryTime()) > (Math.random() * RESEND_PACKET_INTERVAL)) {
                //resend packet if we did not receive ack
                packet.setLastRetryTime(System.currentTimeMillis());
                packet.setRetryCount(packet.getRetryCount() + 1);
                if (packet.getRetryCount() > MAX_RETRY_COUNT) {
                    if (messageNotSentEvent != null) {
                        try {
                            iter.remove();
                            messageNotSentEvent.failure(packet);
                        } catch (Exception ex) {
                            ex.printStackTrace();
                        }
                    }
                } else {
                    messages.put(packet.getPacket());
                    resendsReliable++;
                }
            }
        }

        try {
            Thread.sleep(100);
        } catch (Exception ex) {
        }
    }

    private void sendAckResponse(InetAddress address, int port, byte[] buffer) throws InterruptedException {
        //ack response - 4 byte length. highest two bits will be 11b and rest bits will be an index of ack message
        byte[] array = new byte[6];
        array[0] = (byte) (buffer[0] | 0xC0);
        array[1] = buffer[1];
        array[2] = buffer[2];
        array[3] = buffer[3];
        copyShortToArray(buffer, (short) server.getLocalPort(), 4);
        Message message = new Message(array, address, port);
        messages.put(message);
    }

    public static void copyIntToArray(byte[] bytes, int value, int offset) {
        bytes[offset + 3] = (byte) (value & 0xff);
        bytes[offset + 2] = (byte) ((value >> 8) & 0xff);
        bytes[offset + 1] = (byte) ((value >> 16) & 0xff);
        bytes[offset + 0] = (byte) ((value >> 24) & 0xff);
    }

    public static int copyIntFromArray(byte[] bytes, int offset) {
        int value;
        value = (int) (bytes[0 + offset] & 0xFF);
        value = value << 8;
        value = value | (int) (bytes[1 + offset] & 0xFF);
        value = value << 8;
        value = value | (int) (bytes[2 + offset] & 0xFF);
        value = value << 8;
        value = value | (int) (bytes[3 + offset] & 0xFF);
        return value;
    }

    protected static void copyShortToArray(byte[] bytes, short value, int offset) {
        bytes[offset + 1] = (byte) ((value) & 0xff);
        bytes[offset + 0] = (byte) ((value >> 8) & 0xff);
    }

    protected static short copyShortFromArray(byte[] bytes, int offset) {
        int value;
        value = (int) (bytes[0 + offset] & 0xFF);
        value = value << 8;
        value = value | (int) (bytes[1 + offset] & 0xFF);
        return (short) value;
    }
}
